package com.lanmao.emc.dao.impl;

import com.alibaba.fastjson2.JSON;
import com.lanmao.emc.dao.UpgradeDao;
import com.lanmao.emc.entity.FeIndexLogPO;
import com.lanmao.emc.entity.FeVersionPO;
import com.lanmao.emc.entity.VersionInfo;
import com.lanmao.emc.entity.VersionUsedInfo;
import com.lanmao.emc.enums.DBType;
import com.lanmao.emc.utils.EsUtils;
import com.lanmao.emc.utils.QueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.TopHits;
import org.elasticsearch.search.aggregations.metrics.TopHitsAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.stereotype.Repository;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;

import static com.lanmao.emc.constant.FieldConstant.*;
import static com.lanmao.emc.utils.UpgradeUtil.getEsTable;

/**
 * 描述：
 *
 * @author lanmao
 * @since 2024年11月12日 20:41
 */
@Repository
@Slf4j
public class EsUpgradeDaoImpl implements UpgradeDao {

    @Override
    public boolean allTableExist(String... tables) {
        for (String table : tables) {
            try {
                if (!EsUtils.existsIndex(getEsTable(table))) {
                    return false;
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
        return true;
    }

    @Override
    public FeVersionPO findLastVersion() {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.size(1).sort("id", SortOrder.DESC);
        SearchResponse search = EsUtils.search(sourceBuilder, getEsTable(TABLE_VERSION));
        if (search.getHits().getHits().length == 0) {
            return null;
        }
        return JSON.parseObject(search.getHits().getHits()[0].getSourceAsString(), FeVersionPO.class);
    }

    @Override
    public List<VersionUsedInfo> findAllVersionByIndexName(String indexName) {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder queryBuilder = QueryWrapper.<FeIndexLogPO>builder()
                .filter(FeIndexLogPO::getSuccess, "t")
                .filter(FeIndexLogPO::getIndexName, indexName).build();
        sourceBuilder.query(queryBuilder).sort(CREATE_TIME, SortOrder.DESC).size(10000);
        sourceBuilder.fetchSource(new String[]{INDEX_REL_NAME, IS_USED}, null);
        SearchResponse search = EsUtils.search(sourceBuilder, getEsTable(TABLE_INDEX_LOG));
        return EsUtils.hitsToList(search.getHits().getHits(), VersionUsedInfo.class);
    }

    @Override
    public void batchSave(List<FeIndexLogPO> list) {
        try {
            EsUtils.bulk(list, getEsTable(TABLE_INDEX_LOG));
        } catch (IOException e) {
            log.error("save {} records error,", getEsTable(TABLE_INDEX_LOG), e);
        }
    }

    @Override
    public void batchupdate(List<String> list) {

    }

    @Override
    public void updateIsUsedByIndexAndVersion(String index, String version) {
        UpdateByQueryRequest request = new UpdateByQueryRequest(TABLE_INDEX_LOG);

        // 设置查询条件，这里等同于SQL中的WHERE INDEX_NAME = 'index'
        request.setQuery(QueryBuilders.termQuery(INDEX_NAME, index));

        // 使用脚本设置新的值，这里等同于SQL中的SET IS_USED = CASE WHEN VERSION = 'version' THEN 1 ELSE 0 END
        Script script = new Script(ScriptType.INLINE,
                "painless",
                "if (ctx._source." + VERSION + " == '" + version + "') { ctx._source." + IS_USED + " = 1 } else { ctx._source." + IS_USED + " = 0 }",
                Collections.emptyMap());

        request.setScript(script);
        try {
            EsUtils.updateByQuery(request);
        } catch (IOException e) {
            log.error("update table {} error,", getEsTable(TABLE_INDEX_LOG), e);
        }
    }

    @Override
    public void updateIsUsedByIndexName(List<String> indexList) {
        UpdateByQueryRequest request = new UpdateByQueryRequest(getEsTable(TABLE_INDEX_LOG));

        // 设置查询条件
        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.termQuery(IS_USED, 1))
                .must(QueryBuilders.termsQuery(INDEX_NAME, indexList));

        request.setQuery(queryBuilder);
        request.setRefresh(true);
        // 使用脚本设置新的值
        Script script = new Script(ScriptType.INLINE,
                "painless",
                "ctx._source." + IS_USED + " = 0",
                Collections.emptyMap());

        request.setScript(script);
        try {
            EsUtils.updateByQuery(request);
        } catch (IOException e) {
            log.error("update table {} error,", getEsTable(TABLE_INDEX_LOG), e);

        }
    }

    @Override
    public void deleteByIndexRelName(List<String> indexList) {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder queryBuilder = QueryWrapper.<FeIndexLogPO>builder().terms(FeIndexLogPO::getIndexRelName, indexList.toArray(new String[0])).build();
        sourceBuilder.fetchSource("_id", null);
        sourceBuilder.query(queryBuilder);
        try {
            EsUtils.deleteByQuery(sourceBuilder, getEsTable(TABLE_INDEX_LOG));
        } catch (IOException e) {
            log.error("{} delete records error,", getEsTable(TABLE_INDEX_LOG), e);
        }
    }

    @Override
    public List<VersionInfo> findAllLastErrorIndex() {
        //从库中查询出根据索引名分组的并且根据版本、创建时间以此降序排序的第一条，并找出其中失败的索引
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.size(0);
        TermsAggregationBuilder termsAggBuilder = AggregationBuilders.terms(INDEX_NAME)
                .field(INDEX_NAME)
                .size(10000); // 可以根据需要调整大小
        TopHitsAggregationBuilder topHitsAggBuilder = AggregationBuilders.topHits("top_hits")
                .size(1)
                .sort(VERSION, SortOrder.DESC)
                .sort(CREATE_TIME, SortOrder.DESC)
                .fetchSource(new String[]{INDEX_NAME, VERSION, SUCCESS}, null);
        termsAggBuilder.subAggregation(topHitsAggBuilder);
        sourceBuilder.aggregation(termsAggBuilder);
        SearchResponse response = EsUtils.search(sourceBuilder, getEsTable(TABLE_INDEX_LOG));
        // 处理响应
        Terms aggregation = response.getAggregations().get(INDEX_NAME);
        List<VersionInfo> list = new ArrayList<>();
        for (Terms.Bucket bucket : aggregation.getBuckets()) {
            TopHits topHits = bucket.getAggregations().get("top_hits");
            for (SearchHit hit : topHits.getHits().getHits()) {
                Map<String, Object> sourceMap = hit.getSourceAsMap();
                if ("f".equals(sourceMap.get(SUCCESS).toString())) {
                    list.add(new VersionInfo(sourceMap.get(VERSION).toString(), sourceMap.get(INDEX_NAME).toString()));
                }
            }
        }
        return list;
    }

    @Override
    public void saveVersion(FeVersionPO po) {
        try {
            EsUtils.save(po, getEsTable(TABLE_VERSION));
        } catch (IOException e) {
            log.error("save version:{} info error,", po.getVersion(), e);
        }
    }

    @Override
    public List<FeVersionPO> findAllVersion() {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.size(10000);
        sourceBuilder.fetchSource(new String[]{REMARK, VERSION}, null);
        SearchResponse search = EsUtils.search(sourceBuilder, getEsTable(TABLE_VERSION));
        return EsUtils.hitsToList(search.getHits().getHits(), FeVersionPO.class);
    }

    @Override
    public DBType getDbType() {
        return DBType.ES;
    }
}